package com.cloud.job.consumer;


import com.cloud.job.doris.DorisStreamLoad;
import com.cloud.job.doris.VehicleMessageDaily;
import com.cloud.job.utils.JsonUtils;
import com.fasterxml.jackson.databind.node.ObjectNode;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

import static com.cloud.job.utils.JsonUtils.*;

/**
 * @author 杨斌
 * 2024-12-2024-12-18
 */

@Slf4j
@Component
@ConfigurationProperties
public class MessageDailyConsumer {
    @Resource
    private DorisStreamLoad dorisStreamLoad;

    @KafkaListener(id="${listener.consumer4.id}",topics = "#{'${listener.consumer4.topics}'.split(',')}", groupId = "${listener.consumer4.groupId}",concurrency = "${listener.consumer4.concurrency}")
    public void processMessage(@Payload List<ConsumerRecord<String, String>> consumerRecords,
                               Acknowledgment acknowledgment) {

        log.info("Starting to process daily messages...");
        List<VehicleMessageDaily> vehicleDailyList = new ArrayList<>();
        try  {
            log.info("Polled {} daily messages ",consumerRecords.size());
            if (consumerRecords.isEmpty()) {
                log.info("No daily messages to process");
                return; // 返回空列表
            }
            for (ConsumerRecord<String, String> record : consumerRecords) {
                try {
                    VehicleMessageDaily vehicleDaily = parseMessage(record.value());
                    vehicleDailyList.add(vehicleDaily);
                } catch (Exception e) {
                    log.error("Error parsing daily messages: {}", record.value(), e);
                }
            }
            if (!vehicleDailyList.isEmpty()) {
                String jsonString = JsonUtils.writeValueAsString(vehicleDailyList);
                long loadStartTime = System.currentTimeMillis();
                dorisStreamLoad.loadJson(jsonString, VehicleMessageDaily.class);
                long loadEndTime = System.currentTimeMillis();
                log.info("Inserted daily messages {} records into database in {} ms", vehicleDailyList.size(), (loadEndTime - loadStartTime));
                // 提交偏移量
                acknowledgment.acknowledge();
                log.info("Committed offsets for processed daily messages.");
            } else {
                log.info("No parsed daily messages to insert into database");
            }
        } catch (Exception e) {
            log.error("Error daily processing real messages", e);
        } finally {
            log.info("Finished daily processing real messages.");
        }
    }

    private VehicleMessageDaily parseMessage(String message) throws Exception {
        ObjectNode objectNode = JsonUtils.readObjectNode(message);
        VehicleMessageDaily.VehicleMessageDailyBuilder vehicleDailyBuilder = VehicleMessageDaily.builder();

        String opTime = getTextValue(objectNode, "dataContent", "localDateTime");
        vehicleDailyBuilder
                .opTime(opTime)
                .vin(getTextValue(objectNode, "header", "vin"))
                .accumulativeMileage(getLongValue(objectNode, "dataContent", "evData", "vehicleData", "accumulativeMileage"))
                .soc(getShortValue(objectNode, "dataContent", "evData", "vehicleData", "soc"))
                .commandType(getTextValue(objectNode, "header", "commandType"))
                .lat(getLongValue(objectNode, "dataContent", "evData", "locationData", "latitude"))
                .latitudeDataType(getTextValue(objectNode, "dataContent", "evData", "locationData", "latitudeDataType"))
                .lng(getLongValue(objectNode, "dataContent", "evData", "locationData", "longitude"))
                .longitudeDataType(getTextValue(objectNode, "dataContent", "evData", "locationData", "longitudeDataType"))
                .valid(getBooleanValue(objectNode, "dataContent", "evData", "locationData", "valid"))
                .status(getShortValue(objectNode, "dataContent", "evData", "locationData", "status"))
                .lockVehicleState(getShortValue(objectNode, "dataContent", "evData", "lockVehicleState", "lockVehicleState"))
                .supportLock(getIntValue(objectNode, "dataContent", "evData", "lockVehicleState", "supportLock"))
                .commandId(getTextValue(objectNode, "dataContent", "evData", "lockVehicleState", "commandId"))
                .receiveTime(getTextValue(objectNode, "receive_time"))
                .vehicleType(getTextValue(objectNode, "vehicle_type"))

                // 新增自定义属性
                .customDataLength(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "customDataLength"))
                .st86Version(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "st86Version"))
                .stm32Version(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "stm32Version"))
                .deviceVersion(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "deviceVersion"))
                .userCode(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "userCode"))
                .vehicleCode(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "vehicleCode"))
                .lockStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "lockStatus"))
                .canStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "canStatus"))
                .antennaStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "antennaStatus"))
                .powerOffStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "powerOffStatus"))
                .sdCardStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "sdCardStatus"))
                .drivingDirection(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "drivingDirection"))
                .lockEnableStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "lockEnableStatus"))
                .chargingCycleCount(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "chargingCycleCount"))
                .accumulatedChargingEnergy(getLongValue(objectNode, "dataContent", "evData", "customMessageData", "accumulatedChargingEnergy"))
                .accumulatedWorkingTime(getLongValue(objectNode, "dataContent", "evData", "customMessageData", "accumulatedWorkingTime"))
                .accumulatedGpsMileage(getLongValue(objectNode, "dataContent", "evData", "customMessageData", "accumulatedGpsMileage"))
                .totalStorageCapacity(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "totalStorageCapacity"))
                .remainingStorageCapacity(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "remainingStorageCapacity"))
                .gpsSignalStrength(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "gpsSignalStrength"))
                .gsmSignalStrength(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "gsmSignalStrength"))
                .keyLockStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "keyLockStatus"))
                .keyLock(getTextValue(objectNode, "dataContent", "evData", "customMessageData", "key"));
        return vehicleDailyBuilder.build();
    }

}
